with tttt as ( select
                    t.waybill_no  as t2_waybill_no
                   ,t.arrival_site_code
                from (
                    select
                    to_date(t.arrival_time) as dt,
                    arrival_site,
                    arrival_site_code,
                    waybill_no,
                    send_next_type,
                    case when send_next_type=2 then (unix_timestamp(last_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600
                             else (unix_timestamp(send_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600 end diff_time
                    from   jms_dm.dm_center_stop_time_dt t where  t.dt='{{ execution_date | date_add(-2) | cst_ds }}'
                ) t
                where diff_time>0
               group by waybill_no,arrival_site_code

)

insert overwrite table jms_dm.dm_center_stop_time_detail_dt partition ( dt )
 --36
select
to_date(t.arrival_time) as date_time,
waybill_no,
send_shipment_no,
arrival_site as center_name,
arrival_site_code as center_name_code,
arrival_time as center_arrival_time,
send_time as center_send_time,
last_time as center_last_time,
send_next_name,
case when send_next_type=2 then '是' else '否' end next_type,
case when send_next_type=2 then (unix_timestamp(last_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600
     else (unix_timestamp(send_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600 end diff_time,
null as invalid_type,
1 as stop_type,
t1.agent_name,
t1.agent_code,
     dt
from jms_dm.dm_center_stop_time_dt t
left join  jms_dim.dim_network_whole_massage t1 on t.arrival_site_code=t1.code

where t.send_time <> '0000-0000-0000' and t.arrival_time < t.send_time  and t.dt='{{ execution_date | date_add(-2) | cst_ds }}'
and (
        case when send_next_type=2 then (unix_timestamp(last_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600
             else (unix_timestamp(send_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600 end >=36
     and
 case when send_next_type=2 then (unix_timestamp(last_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600
             else (unix_timestamp(send_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600 end  < 48
)

union all
--48
 select
    to_date(t.arrival_time) as date_time,
    waybill_no,
    send_shipment_no,
    arrival_site as center_name,
    arrival_site_code as center_name_code,
    arrival_time as center_arrival_time,
    send_time as center_send_time,
    last_time as center_last_time,
    send_next_name,
    case when send_next_type=2 then '是' else '否' end next_type,
    case when send_next_type=2 then (unix_timestamp(last_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600
         else (unix_timestamp(send_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600 end diff_time,
    null as invalid_type,
    2 as stop_type ,
    t1.agent_name,
    t1.agent_code,
    t.dt
    from jms_dm.dm_center_stop_time_dt t
    left join  jms_dim.dim_network_whole_massage t1 on t.arrival_site_code=t1.code

    where t.send_time <> '0000-0000-0000' and t.arrival_time < t.send_time  and t.dt='{{ execution_date | date_add(-2) | cst_ds }}'
    and ( case when send_next_type=2 then (unix_timestamp(last_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600
         else (unix_timestamp(send_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600 end >=48
       )
--无效
union all 
select
    date_time,
    waybill_no,
    send_shipment_no,
    center_name as center_name,
    center_name_code as center_name_code,
    center_arrival_time as center_arrival_time,
    center_send_time as center_send_time,
    center_last_time as center_last_time,
    send_next_name,
     next_type,
      diff_time,
    invalid_type,
     stop_type ,
    agent_name,
   agent_code,
    dt
 from (
        select
            dt as date_time,
            waybill_no,
            send_shipment_no,
            arrival_site as center_name,
            t.arrival_site_code as center_name_code,
            arrival_time as center_arrival_time,
            send_time as center_send_time,
            last_time as center_last_time,
            send_next_name,
            case when send_next_type=2 then '是' else '否' end next_type,
            null as diff_time,
            case when arrival_time is null then '有发无到'
                 when send_time is null then '有到无发'
                 when (case when send_next_type=2 then (unix_timestamp(last_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600
                         else (unix_timestamp(send_time) - unix_timestamp(nvl(cancal_time,arrival_time))) end  < 0) then  '先发后到'
                 when (case when send_next_type=2 then (unix_timestamp(last_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600
                         else (unix_timestamp(send_time) - unix_timestamp(nvl(cancal_time,arrival_time))) end  is null)
                      and last_time is null then  '发车时间为空'
                         end as invalid_type,
            3 as stop_type ,
            t1.agent_name,
            t1.agent_code,
            dt,
             row_number() over(partition by waybill_no,t.arrival_site_code order by nvl(send_time,nvl(cancal_time,arrival_time)) desc) as rn
            from jms_dm.dm_center_stop_time_dt t
            left join  jms_dim.dim_network_whole_massage t1 on t.arrival_site_code=t1.code
            left join  tttt t2 on t.waybill_no=t2.t2_waybill_no and t2.arrival_site_code=t.arrival_site_code
            where  t.dt='{{ execution_date | date_add(-2) | cst_ds }}'
            and ( arrival_time is null  or send_time is null or
                  (case when send_next_type=2 then (unix_timestamp(last_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600
                         else (unix_timestamp(send_time) - unix_timestamp(nvl(cancal_time,arrival_time))) end  < 0) or

                  (case when send_next_type=2 then (unix_timestamp(last_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600
                         else (unix_timestamp(send_time) - unix_timestamp(nvl(cancal_time,arrival_time))) end  is null ) or

                  (case when send_next_type=2 then (unix_timestamp(last_time) - unix_timestamp(nvl(cancal_time,arrival_time))) / 3600
                         else (unix_timestamp(send_time) - unix_timestamp(nvl(cancal_time,arrival_time))) end  = 0)
                 )
           and t2.t2_waybill_no is null
)  t where t.rn=1  and invalid_type is not null

 distribute by pmod(hash(rand()), 3);